package org.voovan.korla.rpc.service;

import junit.framework.TestCase;
import org.voovan.korla.KorlaStatic;
import org.voovan.korla.rpc.TestObject2;
import org.voovan.korla.rpc.consumer.RpcMethod;
import org.voovan.korla.rpc.exception.RpcCallbackException;
import org.voovan.korla.rpc.message.RpcCall;
import org.voovan.korla.rpc.message.RpcCallback;
import org.voovan.tools.TEnv;
import org.voovan.tools.TObject;
import org.voovan.tools.json.JSON;
import org.voovan.tools.log.Logger;
import org.voovan.tools.serialize.TSerialize;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 类文字命名
 *
 * @author: helyho* korla Framework.
 * WebSite: https://github.com/helyho/korla
 * Licence: Apache v2 License
 */
public class RpcConsumeUnit extends TestCase {

    public void testBuildConsumeClass() throws TimeoutException {
        RpcConsumer consumer = RpcConsumer.newInstance("127.0.0.1", 10331, 5000, 5000, 1, 1);
        consumer.buildConsumeClass("src/test/java");
    }

    public void testSync() throws TimeoutException, RpcCallbackException {
        RpcConsumer consumer = RpcConsumer.newInstance("127.0.0.1", 10331, 5000, 5000, 1, 1);
        System.out.println((String) consumer.call(RpcCall.newInstance("RpcMethod/rpc1", null, 123), 1000));
    }

    public void testCallback() {
        RpcConsumer consumer = RpcConsumer.newInstance("127.0.0.1", 10331, 5000, 5000, 1, 1);
        consumer.call(RpcCall.newInstance("RpcMethod/rpc2", 123, "str"), 1000, new RpcCallback<String>() {
            @Override
            public void onReturn(String returnValue) {
                System.out.println(returnValue);
            }

            @Override
            public void onException(RpcCallbackException e) {
                e.printStackTrace();
            }
        });

        for (int i = 0; i < 100000; i++) {
            TEnv.sleep(1);
        }
    }

    public void testRpc2() throws TimeoutException, RpcCallbackException {
        RpcConsumer consumer = RpcConsumer.newInstance("127.0.0.1", 10331, 5000, 5000, 1, 1);

        RpcMethod rmt = consumer.build(RpcMethod.class);
        rmt.rpc2(789789, "666", new RpcCallback<String>() {
            @Override
            public void onReturn(String returnValue) {
                System.out.println(returnValue);
            }

            @Override
            public void onException(RpcCallbackException e) {
                e.printStackTrace();
            }
        });

        TEnv.sleep(500);
    }


    public void testRpc4() throws TimeoutException, RpcCallbackException {

        RpcMethod rpcMethod = new RpcMethod("127.0.0.1", 10331);

        org.voovan.korla.rpc.TestObject2 testObject2 = new org.voovan.korla.rpc.TestObject2();
        testObject2.setString("rpc4 test");

        Object ret = rpcMethod.rpc4(testObject2, "asdfadfasdf");
        System.out.println(JSON.toJSON(ret));

        TEnv.sleep(5 * 1000);
    }


    public void testRpc5() throws TimeoutException, RpcCallbackException {

        RpcMethod rpcMethod = new RpcMethod("127.0.0.1", 10331, 5000, 5000, 1, 1);

        TestObject2 testObject2 = new TestObject2();
        testObject2.setString("rpc4 test");

        for (int i = 0; i < 10; i++) {
            Object ret = null;
            try {
                ret = rpcMethod.rpc5((ArrayList) TObject.asList(testObject2), null, ("asdfadfasdf-" + i));
            } catch (TimeoutException e) {
                e.printStackTrace();
            } catch (RpcCallbackException e) {
                e.printStackTrace();
            }
            System.out.println(JSON.toJSON(ret));
        }
    }

    public void testBench() throws TimeoutException {
        RpcMethod rpcMethod = new RpcMethod("127.0.0.1", 10331, 5000, 5000, 50, 50);

        AtomicInteger mm = new AtomicInteger(1);
        AtomicInteger kk = new AtomicInteger(1);
        ArrayList<Thread> ts = new ArrayList<>();

        TEnv.measure("cost", () -> {
            for (int p = 0; p < 8; p++) {
                Thread thread = new Thread(() -> {
                    for (int i = 0; i < 80000; i++) {

                        rpcMethod.rpc1(mm.getAndIncrement() + "", "strstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrs", mm.getAndIncrement(), new RpcCallback<String>() {
                            @Override
                            public void onReturn(String returnValue) {
                                Logger.simple(System.currentTimeMillis());
                                Logger.simple(System.currentTimeMillis());
                            }

                            @Override
                            public void onException(RpcCallbackException e) {

                            }
                        });
//                        RpcCall rpcCall = RpcCall.newInstance("RpcMethod/rpc1", "strstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrstrs", mm.getAndIncrement());
//                        //设置用于分片的key
//                        rpcCall.setSliceKey(rpcCall.getId());
//
//                        consumer.call(rpcCall, new RpcCallback<String>() {
//                                    @Override
//                                    public void onReturn(String returnValue) {
//                                        kk.getAndIncrement();
//                                        Logger.simple(System.currentTimeMillis());
//                                    }
//
//                                    @Override
//                                    public void onException(RpcCallbackException e) {
//                                        e.printStackTrace();
//                                    }
//                                });

                    }

                    TEnv.sleep(2000);
                    System.out.println("done " + mm.get() + " " + kk.get());

                });

                ts.add(thread);
                thread.start();
            }

            for (Thread t : ts) {
                try {
                    t.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });


        int m = 0;
        while (m < 10000000) {
            TEnv.sleep(1000);
            m++;
            if (m % 1000 == 0) {
                System.out.println("Cache size: " + KorlaStatic.MESSAGE.size() + ", Timeout size: " + KorlaStatic.TIMEOUT_MSG.size());
            }

            if (m % 10 == 0) {
                if (mm.get() == kk.get()) {
                    System.out.println("done " + mm.get() + " " + kk.get());
                    break;
                }
            }
        }
    }

    public void testRetry() throws TimeoutException {
        RpcConsumer consumer = RpcConsumer.newInstance("127.0.0.1", 10331, 5000, 5000, 80, 80);

        AtomicInteger mm = new AtomicInteger(1);
        ArrayList<Thread> ts = new ArrayList<>();

        for (int p = 0; p < 4; p++) {
            Thread thread = new Thread(() -> {
                for (int i = 0; i < 20000; i++) {
//                    if(i%300==0) {
//                        TEnv.sleep(1);
//                    }

                    consumer.call(RpcCall.newInstance("RpcMethod/retry", "str", mm.getAndIncrement()),
                            new RpcCallback<String>() {
                        @Override
                        public void onReturn(String returnValue) {
                            System.out.println(returnValue + " " + System.currentTimeMillis());
                        }

                        @Override
                        public void onException(RpcCallbackException e) {
                            e.printStackTrace();
                        }
                    });

                }
                System.out.println("done " + mm.get());

            });

            ts.add(thread);
            thread.start();
        }

        for (Thread t : ts) {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        int m = 0;
        while (m < 10000000) {
            TEnv.sleep(1);
            m++;
            if (m % 1000 == 0) {
                System.out.println("Cache size: " + KorlaStatic.MESSAGE.size() + ", Timeout size: " + KorlaStatic.TIMEOUT_MSG.size());
            }
        }
    }

    public static void main(String[] args) {
        byte[] bbb = TSerialize.serialize(new BigDecimal("123123"));
        System.out.println(TSerialize.unserialize(bbb));
    }
}
